package com.bigdata.flink.datastream;

import com.bigdata.flink.util.StreamEnvUtil;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

// 侧输出流、分流，把不符合条件的数据输出到其它流里面去
public class SideOutputFunctionTest {
  public static void main(String[] args) throws Exception {
    sideOutPut1();
  }

  public static void sideOutPut1() throws Exception {
    StreamExecutionEnvironment env = StreamEnvUtil.getEnv();
    DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

    OutputTag<String> java = new OutputTag<>("java", Types.STRING);
    OutputTag<String> flink = new OutputTag<>("flink", Types.STRING);

    SingleOutputStreamOperator<String> operator = socketSource.process(new ProcessFunction<String, String>() {
      @Override
      public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        if (value.contains("java")) {
          ctx.output(java, value);
        } else if (value.contains("flink")) {
          ctx.output(flink, value);
        } else {
          out.collect(value);
        }
      }
    });

    operator.getSideOutput(java).print("java");
    operator.getSideOutput(flink).print("flink");
    operator.print("other");
    env.execute();
  }

  public static void filter() throws Exception {
    StreamExecutionEnvironment env = StreamEnvUtil.getEnv();
    DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

    // 并行度 1，看下分流后 怎么处理?
    env.setParallelism(2);

    socketSource.filter(s -> Integer.parseInt(s) % 2 == 0).print();
    socketSource.filter(s -> Integer.parseInt(s) % 2 == 1).print();

    env.execute();
  }
}
